package com.coolingme.consumer.auto;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 消费者客户端，自动提交offset
 * <p>
 * Consumer 消费数据时的可靠性是很容易保证的，因为数据在 Kafka 中是持久化的，故不用担心数据丢失问题。
 * 由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置的继续消费，
 * 所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。
 * 所以 offset 的维护是 Consumer 消费数据是必须考虑的问题
 * <p>
 * 屏蔽kafka debug
 * <logger name="org.apache.kafka" level="off"/>
 *
 * @author wangyue
 * @date 2021/7/15 11:04
 */
public class ConsumerDemo {

    public static void main(String[] args) {

        // 配置信息
        Properties properties = new Properties();
        // 连接的集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
//        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "139.196.177.218:9092");
        // 开启自动提交
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, true);
        // 自动提交的延时,ms
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // key,value反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        // 更换一个新的消费者组
//        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");

        // 重置消费者的offset
        // 当前消费者组不存在或服务上没有数据（如：数据被删除）
        // earliest:重置为最早的offset，获取最早的数据
        // latest:重置为最新的offset，获取最新的数据
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 消费者
        KafkaConsumer<String, Object> consumer = new KafkaConsumer<>(properties);
        // 订阅主题,若主题不存在会自动创建
        consumer.subscribe(Arrays.asList("test-topic", "coolingme"));
        while (true) {
            // 拉取数据
            ConsumerRecords<String, Object> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, Object> record : records) {
                System.out.println(String.format("key= %s ,value= %s ,topic= %s ,partition= %d ,offset= %d", record.key(), record.value(), record.topic(), record.partition(), record.offset()));
            }
        }

    }
}

